package com.ln.kafka.v2_4_0.stream;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Produced;

import java.util.Arrays;
import java.util.Locale;
import java.util.Properties;

/**
 * <pre>
 *     测试流：
 *     命令行往队列里面放数据：
        bin/kafka-console-producer.sh --broker-list localhost:9092 --topic topic-stream-in
 *     命令行从队列里面取数据:
        bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
            --topic topic-stream-out \
            --property print.key=true \
            --property print.value=true \
            --property key.deserializer=org.apache.kafka.common.serialization.StringDeserializer \
            --property value.deserializer=org.apache.kafka.common.serialization.StringDeserializer \
            --from-beginning


 bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
 --topic topic-stream-out
 --from-beginning
 *
 * </pre>
 *
 * @Author zhangj
 * @Date 2022/6/8 21:33
 */
public class StreamSimple {

    public static String INPUT_TOPIC_NAME = "topic-stream-in";
    public static String OUTPUT_TOPIC_NAME = "topic-stream-out";

    public static void main(String[] args) {
        Properties props = new Properties();
        props.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-app");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        // 构建流结构拓扑
        final StreamsBuilder builder = new StreamsBuilder();
        // 构建workcount processccc
//        wordCountStream(builder);

        foreachStream(builder);
        KafkaStreams streams = new KafkaStreams(builder.build(), props);

        streams.start();

    }

    /**
     * 定义流计算过程
     */
    static void wordCountStream(StreamsBuilder builder) {
        // 读取源数据：不断从inout topic上获取新数据，并且追加到流上的一个抽象对象
        KStream<String, String> source = builder.stream(INPUT_TOPIC_NAME);
        // Hello World  : 按照空格截取获取单词数量
        // KTable 是数据集合的抽象对象
        // 引出概念：算子
        KTable<String, Long> count = source
                // flatMapValues ： 将一行数据拆分为多行数据 key 1, value Hello World
                // flatMapValues ： 将一行数据拆分为多行数据 key 1, value Hello key xx, value World
                .flatMapValues(val -> Arrays.asList(val.toLowerCase(Locale.getDefault()).split(" ")))
                // groupBy: 按照 value 值来合并
                .groupBy((key, val) -> val)
                // 统计出现的总数
                .count();

        count.toStream().to(OUTPUT_TOPIC_NAME, Produced.with(Serdes.String(), Serdes.Long()));
    }

    /**
     * 演示 flatMap 的数据
     */
    static void foreachStream(StreamsBuilder builder) {
        KStream<String, String> source = builder.stream(INPUT_TOPIC_NAME);
        source
                // flatMapValues ： 将一行数据拆分为多行数据 key 1, value Hello World
                // flatMapValues ： 将一行数据拆分为多行数据 key 1, value Hello key xx, value World
                .flatMapValues(val -> Arrays.asList(val.toLowerCase(Locale.getDefault()).split(" ")))
                .foreach(((key, value) -> System.out.println("key: " + key + " ; value: " + value)));
    }

}
